"""
Case Type   : 防篡改
Case Name   : 账本数据库基础语法适配(DDL/DML)--enable_copy_server_files打开,使用copyto命令导出防篡改用户表
Create At   : 2022-02-28
Owner       : lwx1080719
Description :
    1.创建存储防篡改用户表的文件
    2.查询enableSeparationOfDuty和enable_copy_server_files
    3.修改enable_copy_server_files为on
    4.创建不同权限用户
    5.系统管理员创建schema并指定防篡改选项
    6.普通用户登录数据库在防篡改schema下创建防篡改用户表并插入数据
    7.不同权限用户登录数据库使用copyto命令导出防篡改用户表
    8.清理环境
Expect      :
    1.创建成功
    2.均显示默认设置off
    3.成功
    4.成功
    5.成功
    6.成功
    7.系统管理员导出成功，其他用户导出失败
    8.成功
History     :
"""




import os
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Constant import Constant
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Common import Common
from testcase.utils.Logger import Logger


class ModifyCase(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f'-----{os.path.basename(__file__)} start-----')
        self.dbuserNode = Node('PrimaryDbUser')
        self.primary_sh = CommonSH('PrimaryDbUser')
        self.Constant = Constant()
        self.Common = Common()
        self.schema_name = 's_security_tamper_proof_0043'
        self.table_name = 't_security_tamper_proof_0043'
        self.user_file = 'user.bat'
        self.file_path = os.path.join(macro.DB_INSTANCE_PATH,
                                      self.user_file)
        self.user_name = 'u_security_tamper_proof'
        self.user_privileges = {f'{self.user_name}_sys': 'sysadmin',
                                f'{self.user_name}_create': 'createrole',
                                f'{self.user_name}_audit': 'auditadmin',
                                f'{self.user_name}_noruser': '',
                                f'{self.user_name}_independent': 'independent',
                                f'{self.user_name}_mon': 'monadmin',
                                f'{self.user_name}_opr': 'opradmin',
                                f'{self.user_name}_pol': 'poladmin',
                                f'{self.user_name}_persistent': 'persistence'}
        self.default_value1 = self.Common.show_param('enableSeparationOfDuty')
        self.default_value = self.Common.show_param('enable_copy_server_files')

    def test_security(self):
        text = '----step1:创建存储防篡改用户表的文件 expect:成功----'
        self.logger.info(text)
        cmd = f'''touch {self.file_path};
                ls {macro.DB_INSTANCE_PATH} | grep {self.user_file};'''
        self.logger.info(cmd)
        result = self.Common.get_sh_result(self.dbuserNode, cmd)
        self.assertIn(self.user_file, result, "创建成功" + text)

        text = '----step2:查询参数值  expect:默认值off----'
        self.logger.info(text)
        self.assertEqual("off", self.default_value, "执行失败:" + text)
        self.assertEqual("off", self.default_value1, "执行失败:" + text)

        text = '----step3:修改enable_copy_server_files为on ' \
               'expect:成功----'
        self.logger.info(text)
        sql_cmd = self.primary_sh.execut_db_sql(f'''
                    alter system set enable_copy_server_files to on;
                    select pg_sleep(2);
                    show enable_copy_server_files;''')
        self.logger.info(sql_cmd)
        self.assertEqual("on", sql_cmd.split("\n")[-2].strip(),
                         "执行失败:" + text)

        text = '----step4:创建不同权限用户 expect:成功----'
        self.logger.info(text)
        for user, privilege in self.user_privileges.items():
            create_cmd = f"drop user if exists {user} cascade; " \
                f"create user {user} {privilege} " \
                f"password '{macro.PASSWD_REPLACE}' ; "
            msg = self.primary_sh.execut_db_sql(create_cmd)
            self.logger.info(msg)
            self.assertIn(self.Constant.DROP_ROLE_SUCCESS_MSG and
                          self.Constant.CREATE_ROLE_SUCCESS_MSG, msg,
                          '执行失败:' + text)

        text = '----step4:系统管理员创建schema并指定防篡改选项 expect:成功----'
        self.logger.info(text)
        sql_cmd = f'''
            create schema {self.schema_name} authorization
            {self.user_name}_noruser with blockchain;'''
        self.logger.info(sql_cmd)
        msg = self.primary_sh.execut_db_sql(
            sql_cmd, sql_type=f' -U {self.user_name}_sys '
            f'-W {macro.PASSWD_REPLACE}')
        self.logger.info(msg)
        self.assertTrue(
            msg.count(self.Constant.CREATE_SCHEMA_SUCCESS_MSG) == 1,
            "执行失败:" + text)

        text = '----step5:普通用户登录数据库在防篡改schema下' \
               '创建防篡改用户表并插入数据 expect:成功----'
        self.logger.info(text)
        sql_cmd = f'''
            create table {self.schema_name}.{self.table_name}
            (id int primary key ,name varchar(20) NOT NULL ,
            address text DEFAULT 'testui');
            insert into {self.schema_name}.{self.table_name}
            values(1,'swf',DEFAULT),(2,'yyq','xiaomi');'''
        self.logger.info(sql_cmd)
        msg = self.primary_sh.execut_db_sql(
            sql_cmd, sql_type=f' -U {self.user_name}_noruser '
            f'-W {macro.PASSWD_REPLACE}')
        self.logger.info(msg)
        self.assertIn(self.Constant.CREATE_TABLE_SUCCESS and
                      self.Constant.INSERT_SUCCESS_MSG, msg,
                      "执行失败:" + text)

        text = '----step6:不同权限用户登录数据库使用' \
               'copyto命令导出防篡改用户表 expect:系统管理员导出成功,' \
               '其他用户导出失败----'
        self.logger.info(text)
        sql_cmd = f'''copy {self.schema_name}.{self.table_name} to 
                    '{self.file_path}';'''
        self.logger.info(sql_cmd)
        for user in self.user_privileges.keys():
            msg = self.primary_sh.execut_db_sql(
                sql_cmd, sql_type=f' -U {user} -W {macro.PASSWD_REPLACE}')
            self.logger.info(msg)
            if user in {f'{self.user_name}_sys'}:
                self.assertIn('COPY', msg, '执行失败:' + text)
            else:
                self.assertIn(f'must be system admin or a member ', msg,
                              "执行失败:" + text)

    def tearDown(self):
        text = '----step7:清理环境 expect:成功----'
        self.logger.info(text)
        sql_cmd1 = self.primary_sh.execut_db_sql(f'''
            drop table  {self.schema_name}.{self.table_name} cascade;
            drop schema {self.schema_name} cascade;''')
        self.logger.info(sql_cmd1)
        text = '----删除用户 expect:成功----'
        self.logger.info(text)
        drop_msg = []
        for user in self.user_privileges.keys():
            drop_cmd = f"drop user if exists {user} cascade; "
            msg = self.primary_sh.execut_db_sql(drop_cmd)
            self.logger.info(msg)
            drop_msg.append(msg)
        text = '-----恢复参数enable_copy_server_files默认值-----'
        self.logger.info(text)
        sql_cmd2 = self.primary_sh.execut_db_sql(f'''
            alter system set enable_copy_server_files to {self.default_value};
            select pg_sleep(2);
            show enable_copy_server_files;''')
        self.logger.info(sql_cmd2)
        self.assertEqual("off", sql_cmd2.split("\n")[-2].strip(),
                         "执行失败:" + text)
        self.assertIn(self.Constant.DROP_SCHEMA_SUCCESS_MSG and
                      self.Constant.DROP_TABLE_SUCCESS, sql_cmd1,
                      "执行失败:" + text)
        self.assertEqual(9,
                         drop_msg.count(self.Constant.DROP_ROLE_SUCCESS_MSG),
                         '执行失败:' + text)
        text = '-----删除文件-----'
        self.logger.info(text)
        del_file = f'''rm -rf {self.file_path}'''
        self.logger.info(del_file)
        self.Common.get_sh_result(self.dbuserNode, del_file)
        cmd = f'if [ -f {self.file_path} ]; then echo "does exists"; ' \
            f'else echo "not exists"; fi'
        self.logger.info(del_file)
        file_judgments = self.Common.get_sh_result(self.dbuserNode, cmd)
        self.assertIn('not exists', file_judgments, "执行失败" + text)
        self.logger.info(f'-----{os.path.basename(__file__)} end-----')
